from application.etl.base_etl import BaseEtl
from application.etl.write.print_write import PrintWriter
from application.pipelines.information_deduplication_pipeline import InformationDeduplicationPipeline

from application.etl.write.kafka_to_info_write import KafkaToInfoWrite
from application.models.kafka_pathway_schema import KafkaPathwaySchema
from application.settings import KAFKA_CONFIG
import pathway as pw


class KafkaToInfo(BaseEtl):
    """
    Kafka → 资源库信息表 ETL 类。

    该类的主要职责：
    1. 从 Kafka 消费信息数据；
    2. 应用信息去重等管道处理逻辑；
    3. 将处理结果写入资源库信息表（或其他观察者）。

    特点：
    - 使用 pathway 的 Kafka IO 模块读取数据流；
    - 支持批量写入与时间窗口配置。
    """

    # 预定义处理管道（信息去重）
    pipelines = [InformationDeduplicationPipeline()]
    topic_name = 'temp5'

    def read(self):
        """
        从 Kafka 主题读取信息数据。

        配置项来源于 ``KAFKA_CONFIG['pathway_consumer']``，读取 JSON 格式消息，
        并使用 ``KafkaPathwaySchema`` 进行字段解析和校验。

        :return: pathway.Table，包含 Kafka 消费到的消息
        """
        input_data = pw.io.kafka.read(
            KAFKA_CONFIG["pathway_consumer"],
            topic=self.topic_name,  # Kafka 主题
            format="json",  # 消息格式
            schema=KafkaPathwaySchema,  # 消息 schema
            autocommit_duration_ms=1000,  # 自动提交间隔
            max_backlog_size=500,  # 消息积压上限
        )
        return input_data

    def write(self, transform_data):
        """
        将 transform 阶段处理后的数据写入目标。

        当前实现写入资源库（通过 ``KafkaToInfoWrite``），支持批量大小和时间窗口参数。

        :param transform_data: pathway.Table，经 transform 处理后的结果
        :return: None
        """
        # 调试时可使用 PrintWriter 输出到控制台
        # pw.io.python.write(transform_data, PrintWriter())

        # 实际写入 MySQL 资源库（或其他持久化目标）
        pw.io.python.write(transform_data, KafkaToInfoWrite(batch_size=50, time_window=10))
